package com.bungarus.model;

import com.bungarus.busi.Callback;
import com.bungarus.busi.InnerClient;
import com.bungarus.busi.Remedy;
import com.bungarus.busi.ServerLoggingHandler;
import com.bungarus.cluster.*;
import com.bungarus.utils.IPAddressUtils;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import org.apache.log4j.Logger;
import org.springframework.data.mongodb.core.MongoTemplate;

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Stream;

/**
 * Created by tang on 2018/11/24.
 */
abstract public class AbstractBungarusServer implements Node, BungarusServer, BungarusClient {
    final Logger logger = Logger.getLogger(AbstractBungarusServer.class);

    public static final String CLUSTER_TAG_ACCESS = "access";
    public static final String CLUSTER_TAG_TRANSFER = "transfer";
    public static final String CLUSTER_TAG_BUSI = "busi";
    // the busi cluster will dispatch the message to large group processor through the busi bridge node
    public static final String CLUSTER_TAG_BUSI_BRIDGE = "busi-bridge";

    private final NioEventLoopGroup boss = new NioEventLoopGroup();
    private final NioEventLoopGroup worker = new NioEventLoopGroup();
    private final List<Channel> serverChannels = new ArrayList<>(0);
    private final Map<String, InnerClient> innerClients = new HashMap<>(0);

    protected final Map<String, CopyOnWriteArrayList<NodeInfo>> remoteServerNodes = new HashMap<>(0);

    protected final Map<String, RemoteServerAndInnerClientConnectionMap> remoteServerAndInnerClientConnectionMap;

    protected static final String HOST = "host";

    private NodeInfo nodeInfo;
    //the address of the binded host
    private String bindHost;
    private int status = Node.STATUS_STOP;
    private final ClusterManager clusterManager = new ClusterManager(getTag());

    /**
     * get current tag of server
     */
    abstract public String getTag();


    /**
     * channel handler initializer for handling request
     * @return
     */
    abstract public ChannelHandler getChildHandlerInitializer();

    /**
     * return the tags of related clusters
     * @return
     */
    abstract public String[] getRemoteNodeTags();

    /**
     * mongo template (spring data concept) for storing failed retry messages
     * @return mongo template
     */
    abstract protected MongoTemplate getMongoTemplate4FailedRetry();

    public AbstractBungarusServer() {
        remoteServerAndInnerClientConnectionMap = new HashMap<>();
        Stream.of(getRemoteNodeTags()).forEach(remoteServerTag -> remoteServerAndInnerClientConnectionMap.put(remoteServerTag, new RemoteServerAndInnerClientConnectionMap()));

        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
        scheduledExecutorService.scheduleAtFixedRate(() -> {
            for(String key : remoteServerAndInnerClientConnectionMap.keySet()) {
                System.out.println(String.format("In cluster %s: ", key));
                Enumeration<IPAddressUtils.HostPortPair> servers = remoteServerAndInnerClientConnectionMap.get(key).connectionMap.keys();
                System.out.println(String.format("%-10s", "Remote server length: " + remoteServerAndInnerClientConnectionMap.get(key).connectionMap.keySet().size()));
                while (servers.hasMoreElements()) {
                    IPAddressUtils.HostPortPair hostPortPair = servers.nextElement();
                    System.out.println(String.format("%-10s", "Remote server: " + hostPortPair.host + ":" + hostPortPair.port +  " (hash code: " + hostPortPair.hashCode() + ") was connected by: "));
                    CopyOnWriteArrayList<InnerClient> value = remoteServerAndInnerClientConnectionMap.get(key).connectionMap.get(hostPortPair);
                    for (InnerClient client : value) {
                        System.out.println(String.format("%-20s%s", " ", client.toString() + ";"));
                    }
                }
            }
        }, 2, 5, TimeUnit.SECONDS);

    }

    /**
     * the cluster manager MUST ONLY has ONE instance.
     */
    public ClusterManager getClusterManager() {
        return clusterManager;
    }

    public void registerInnerClientConnectionMap(String remoteServerClusterTag, IPAddressUtils.HostPortPair hostPortPair, InnerClient client) {
        this.remoteServerAndInnerClientConnectionMap.get(remoteServerClusterTag).addConnectionMap(hostPortPair.copy(), client);
    }

    public void unregisterInnerClientConnectionMap(String remoteServerClusterTag, IPAddressUtils.HostPortPair hostPortPair, InnerClient client) {
        this.remoteServerAndInnerClientConnectionMap.get(remoteServerClusterTag).removeConnectionMap(hostPortPair.copy(), client);
    }

    public void unregisterInnerClientConnectionMap(final IPAddressUtils.HostPortPair hostPortPair, final InnerClient client) {
        this.remoteServerAndInnerClientConnectionMap.values().forEach(value -> value.removeConnectionMap(hostPortPair.copy(), client));
    }

    @Override
    public String getHostAddress() throws NodeException {
        return bindHost;
    }

    @Override
    public void bind(String serviceHost, int[] servicePorts, int[] heartBeatPorts) throws BungarusServerException {
        if(null == getChildHandlerInitializer()) {
            throw new BungarusServerException("server bind error: no handler.");
        }
        if(null == servicePorts) {
            throw new BungarusServerException("server bind error: no service listening port.");
        }

        ServerBootstrap server = new ServerBootstrap();
        server.group(boss, worker)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 168)
                .handler(new ServerLoggingHandler(LogLevel.DEBUG))
                .childHandler(getChildHandlerInitializer());

        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                if(null != worker) {
                    worker.shutdownGracefully();
                }
                if(null != boss) {
                    boss.shutdownGracefully();
                }
            }
        });

        try {
            String hostAddress = null == serviceHost? IPAddressUtils.getLocalHostLANAddress().getHostAddress() : serviceHost;
            this.bindHost = hostAddress;
            logger.info("Service is bound to the address -> " + this.bindHost);
            for(int port : servicePorts){
                serverChannels.add(server.bind(hostAddress, port).sync().channel());
            }
            if(null != heartBeatPorts) {
                for (int port : heartBeatPorts) {
                    serverChannels.add(server.bind(hostAddress, port).sync().channel());
                }
            }
        }
        catch (Exception e){
            worker.shutdownGracefully();
            boss.shutdownGracefully();
            throw new BungarusServerException(e);
        }
        logger.info("Server initialized successfully.");
    }

    @Override
    public void initRemedy() throws Exception {
        Remedy.start(this);
    }

    /**
     * registers self as a node in cluster with its tag.
     * retrieves all nodes of related cluster respectively,
     * and connects one of those.
     *
     * @throws BungarusServerException
     */
    @Override
    public void initClusterAndConnectRemoteNode() throws BungarusServerException{
        try {
            //register myself into a cluster
            register2Cluster();
            //gather the remote nodes info
            initRemoteNodesInfo();

            //choose a remote server for each tag to connect, and put all servers in queue for backup
            for (String CLUSTER_TAG : getRemoteNodeTags()) {
                int index = this.nodeInfo.getNodeId() % remoteServerNodes.get(CLUSTER_TAG).size();
                CopyOnWriteArrayList<NodeInfo> queue = calcRemoteServerNodes(remoteServerNodes.get(CLUSTER_TAG));
                // just one inner client for each tag
                innerClients.put(CLUSTER_TAG, this.connect(index, queue));
            }
            this.setStatus(Node.STATUS_RUNNING);
        }
        catch (Exception e) {
            throw new BungarusServerException(e);
        }
    }

    /**
     * gather the remote nodes info
     * @throws InterruptedException
     */
    protected void initRemoteNodesInfo() throws InterruptedException{
        boolean again;
        do {
            //get remote server nodes
            getNodesInCluster();
            again = remoteServerNodes.size() != getRemoteNodeTags().length;
            if(again) {
                logger.info("FAILED TO GET CLUSTER NODES!!! Try again...");
            }
            else {
                for(String CLUSTER_TAG : getRemoteNodeTags()) {
                    if(remoteServerNodes.get(CLUSTER_TAG).size() == 0){
                        logger.info("NO NODES FOUND IN CLUSTER " + CLUSTER_TAG + ", Try again...");
                        again = true;
                        break;
                    }
                }
            }
            if(again) {
                Thread.sleep(2000);
            }
        } while (again);
    }

    @Override
    public void listen(Callback<Object, Void> callback) throws BungarusServerException{
        if(this.serverChannels.size() > 0){
            logger.info("Server start and is listening to ...");
            try {
                for (Channel channel : this.serverChannels) {
                    channel.closeFuture().sync();
                }
            }
            catch (Exception e){
                throw new BungarusServerException(e);
            }
            finally {
                if(null != callback) {
                    callback.call(null);
                }
                shutdown();
            }
        }
    }

    @Override
    public void register2Cluster() throws RegisterNodeException {
        this.nodeInfo = getClusterManager().registerNode(this);
    }

    protected void getNodesInCluster() {
        List<ClusterInfo> clusters;
        for (String CLUSTER_TAG : getRemoteNodeTags()) {
            try {
                clusters = getClusterManager().clusters(CLUSTER_TAG);
                if (null != clusters && clusters.size() > 0) {
                    int index = this.nodeInfo.getNodeId() % clusters.size();
                    List<NodeInfo> nodes = getClusterManager().nodes(CLUSTER_TAG, clusters.get(index).getClusterID());
                    remoteServerNodes.put(CLUSTER_TAG, new CopyOnWriteArrayList<>(nodes));
                }
            }
            catch (Exception e) {
                logger.info(e.getMessage());
                e.printStackTrace();
            }
        }
    }

    @Override
    public InnerClient connect(int index, CopyOnWriteArrayList<NodeInfo> queue) throws BungarusClientException {
        InnerClient client = new InnerClient(this, index, queue, getMongoTemplate4FailedRetry());
        client.run(getTag());
        return client;
    }

    @Override
    public Map<String, InnerClient> getInnerClients() {
        return innerClients;
    }

    @Override
    public int getStatus() {
        return this.status;
    }

    @Override
    public void setStatus(int status) throws Exception {
        this.nodeInfo.setStatus(status);
        this.status = status;
        getClusterManager().setNodeData(this.nodeInfo);
    }

    @Override
    public InnerClient getInnerClientConnectingToBackendClusterNode(String clusterTag) {
        return getInnerClients().get(clusterTag);
    }

    @Override
    public int availableProtocol() {
        return Node.PROTOCOL_TCP;
    }

    @Override
    public void shutdown() throws BungarusServerException {
        logger.info("Shutting down...");
        getClusterManager().close();
        worker.shutdownGracefully();
        boss.shutdownGracefully();
    }

    @Override
    public Optional<NodeInfo> getNodeInfo() {
        return Optional.ofNullable(nodeInfo);
    }

    static final class RemoteServerAndInnerClientConnectionMap {
        private Logger logger = Logger.getLogger(RemoteServerAndInnerClientConnectionMap.class);
        private final ConcurrentHashMap<IPAddressUtils.HostPortPair, CopyOnWriteArrayList<InnerClient>> connectionMap = new ConcurrentHashMap<>();

        public RemoteServerAndInnerClientConnectionMap() {
        }

        public void addConnectionMap(IPAddressUtils.HostPortPair remoteServer, InnerClient client) {
            CopyOnWriteArrayList<InnerClient> beforeMap = connectionMap.putIfAbsent(remoteServer, new CopyOnWriteArrayList<>(Collections.singletonList(client)));
            if(null != beforeMap) {
                beforeMap.add(client);
            }
        }

        public void removeConnectionMap(IPAddressUtils.HostPortPair remoteServer, InnerClient client) {
            if(connectionMap.containsKey(remoteServer)) {
                connectionMap.get(remoteServer).remove(client);
            }
        }

        public boolean hasConnection(IPAddressUtils.HostPortPair remoteServer) {
            return connectionMap.containsKey(remoteServer) && (!connectionMap.get(remoteServer).isEmpty());
        }

        public Map<IPAddressUtils.HostPortPair, CopyOnWriteArrayList<InnerClient>> getClientsMapConnectedToTheSameRemoteServer() {
            Map.Entry<IPAddressUtils.HostPortPair, CopyOnWriteArrayList<InnerClient>> entry1 = connectionMap.entrySet().stream().filter(entry -> entry.getValue().size() > 1).findFirst().orElse(null);
            if(null != entry1) {
                Map<IPAddressUtils.HostPortPair, CopyOnWriteArrayList<InnerClient>> map = new HashMap<>();
                map.put(entry1.getKey(), entry1.getValue());
                return map;
            }
            return null;
        }

        public List<InnerClient> getClientsConnectedToTheSameRemoteServer() {
            return connectionMap.values().stream().filter(clients -> clients.size() > 1).findFirst().orElse(null);
        }
    }
}
